/**
 * This file is a part of qloudgen-sec. 
 * You can redistribute qloudgen-sec and/or modify it under the terms of the Lesser GNU General Public License version 3. 
 * qloudgen-sec is distributed WITHOUT ANY WARRANTY. 
 *
 * See the Lesser GNU General Public License for more details at http://www.gnu.org/licenses/. 
 **/
package org.redisson;

import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.redisson.cluster.ClusterConnectionManager;
import org.redisson.core.RMap;

import com.lambdaworks.redis.RedisConnection;

public class ClusterRedissonMap<K, V> extends ClusterRedissonExpirable implements RMap<K, V> {

	ClusterRedissonMap(ClusterConnectionManager connectionManager, String name) {
		super(connectionManager, name);
	}

	@Override
	public int size() {
		return new ClusterCommand<Long>( connectionManager ) {
			@Override
			public Long execute(RedisConnection connection) {
				return connection.hlen( getName() );
			}
		}.run( getName() ).intValue();
	}

	@Override
	public boolean isEmpty() {
		return size() == 0;
	}

	@Override
	public boolean containsKey(final Object key) {
		return new ClusterCommand<Boolean>( connectionManager ) {
			@Override
			public Boolean execute(RedisConnection connection) {
				return connection.hexists( getName() , key );
			}
		}.run( getName() );
	}

	@Override
	public boolean containsValue(final Object value) {
		return new ClusterCommand<Boolean>( connectionManager ) {
			@Override
			public Boolean execute(RedisConnection connection) {
				return connection.hvals( getName() ).contains( value );
			}
		}.run( getName() );
	}

	@Override
	public V get(final Object key) {
		return ( V ) new ClusterCommand<V>( connectionManager ) {
			@Override
			public V execute(RedisConnection connection) {
				return ( V ) connection.hget( getName() , key );
			}
		}.run( getName() );
	}

	@Override
	public V put(final K key, final V value) {
		return ( V ) new ClusterCommand<V>( connectionManager ) {
			@Override
			public V execute(RedisConnection connection) {
				V prev = ( V )connection.hget( getName() , key );
				connection.hset( getName() , key , value );
				return prev;
			}
		}.run( getName() );
	}

	@Override
	public V remove(final Object key) {
		return ( V ) new ClusterCommand<V>( connectionManager ) {
			@Override
			public V execute(RedisConnection connection) {
				V prev = ( V )connection.hget( getName() , key );
				connection.hdel( getName() , key );
				return prev;
			}
		}.run( getName() );
	}

	@Override
	public void putAll(final Map<? extends K, ? extends V> map) {
		new ClusterCommand<String>( connectionManager ) {
			@Override
			public String execute(RedisConnection connection) {
				return connection.hmset( getName() , map );
			}
		}.run( getName() );
	}

	@Override
	public void clear() {
		new ClusterCommand<Long>( connectionManager ) {
			@Override
			public Long execute(RedisConnection connection) {
				return connection.del( getName() );
			}
		}.run( getName() );
	}

	@Override
	public Set<K> keySet() {
		return ( Set<K> ) new ClusterCommand<Set<K>>( connectionManager ) {
			@Override
			public Set<K> execute(RedisConnection connection) {
				return ( Set<K> )connection.hkeys( getName() );
			}
		}.run( getName() );
	}

	@Override
	public Collection<V> values() {
		return ( Collection<V> ) new ClusterCommand<Collection<V>>( connectionManager ) {
			@Override
			public Collection<V> execute(RedisConnection connection) {
				return ( Collection<V> )connection.hvals( getName() );
			}
		}.run( getName() );
	}

	@Override
	public Set<Map.Entry<K, V>> entrySet() {
		return ( Set<Map.Entry<K, V>> ) new ClusterCommand<Set<Map.Entry<K, V>>>( connectionManager ) {
			@Override
			public Set<Map.Entry<K, V>> execute(RedisConnection connection) {
				Map<K, V> map = connection.hgetall( getName() );
				Map<K, V> result = new HashMap<K, V>();
				for ( Map.Entry<K, V> entry : map.entrySet() ) {
					result.put( ( K )entry.getKey() , ( V )entry.getValue() );
				}
				return result.entrySet();
			}
		}.run( getName() );
	}

	@Override
	public V putIfAbsent(final K key, final V value) {
		return ( V ) new ClusterCommand<V>( connectionManager ) {
			@Override
			public V execute(RedisConnection connection) {
				while ( true ) {
					Boolean res = connection.hsetnx( getName() , key , value );
					if ( ! res ) {
						V result = ( V )connection.hget( getName() , key );
						if ( result != null ) {
							return result;
						}
					}
					else {
						return null;
					}
				}
			}
		}.run( getName() );
	}

    private boolean isEquals(RedisConnection<Object, Object> connection, Object key, Object value) {
        Object val = connection.hget(  getName(), key );
        return ( value != null && value.equals( val ) ) || ( value == null && val == null );
    }

	@Override
	public boolean remove(final Object key, final Object value) {
		return new ClusterCommand<Boolean>( connectionManager ) {
			@Override
			public Boolean execute(RedisConnection connection) {
				while ( true ) {
					connection.watch( getName() );
					if ( connection.hexists( getName() , key ) && isEquals( connection , key , value ) ) {
						connection.multi();
						connection.hdel( getName() , key );
						if ( connection.exec().size() == 1 ) {
							return true;
						}
					}
					else {
						connection.unwatch();
						return false;
					}
				}
			}
		}.run( getName() );
	}

	@Override
	public boolean replace(final K key, final V oldValue, final V newValue) {
		return new ClusterCommand<Boolean>( connectionManager ) {
			@Override
			public Boolean execute(RedisConnection connection) {
				while ( true ) {
					connection.watch( getName() );
					if ( connection.hexists( getName() , key ) && isEquals( connection , key , oldValue ) ) {
						connection.multi();
						connection.hset( getName() , key , newValue );
						if ( connection.exec().size() == 1 ) {
							return true;
						}
					}
					else {
						connection.unwatch();
						return false;
					}
				}
			}
		}.run( getName() );
	}

	@Override
	public V replace(final K key, final V value) {
		return ( V ) new ClusterCommand<V>( connectionManager ) {
			@Override
			public V execute(RedisConnection connection) {
				while ( true ) {
					connection.watch( getName() );
					if ( connection.hexists( getName() , key ) ) {
						V prev = ( V )connection.hget( getName() , key );
						connection.multi();
						connection.hset( getName() , key , value );
						if ( connection.exec().size() == 1 ) {
							return prev;
						}
					}
					else {
						connection.unwatch();
					}
					return null;
				}
			}
		}.run( getName() );
	}
}